package com.monitor.zkm.core.mq;

import bean.RocketMqConfig;
import cn.hutool.core.util.StrUtil;
import com.monitor.zkm.core.listener.ZkmMessageListener;
import core.ParamsContext;
import core.annotation.AfterStart;
import core.annotation.StartHandle;
import core.scan.ScanXmlConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

@AfterStart
@Slf4j
@Component
public class RocketMqConsumer implements StartHandle {

    private void initDefaultConsumer(ParamsContext paramsContext) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        RocketMqConfig config = (RocketMqConfig) ScanXmlConfig.newInScanXmlConfig().loadConfig("rocketmq", RocketMqConfig.class);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setVipChannelEnabled(false);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setClientIP(config.getClientIP());
        consumer.setNamesrvAddr(config.getNamesrvAddr());
        consumer.setInstanceName(config.getGroupName());
        consumer.setConsumerGroup(config.getGroupName());
        consumer.setConsumeTimeout(config.getTimeOut());
        consumer.setConsumeMessageBatchMaxSize(config.getMaxMsgSize());
        consumer.registerMessageListener(new ZkmMessageListener(paramsContext));
        String tags = StrUtil.replace(config.getTag(),",","||");
        consumer.subscribe(config.getTopic(),tags);
        consumer.start();
        log.info("consumer start successful");
    }


    @Override
    public void start(ParamsContext paramsContext) throws Exception {
        initDefaultConsumer(paramsContext);
    }
}
